package com.yang.semaphore;

import java.util.concurrent.Semaphore;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.Random;

/**
 * 生产者-消费者模式示例
 * 使用信号量控制生产者和消费者的协调工作
 */
public class ProducerConsumerExample {
    
    // 缓冲区大小
    private static final int BUFFER_SIZE = 5;
    
    // 共享缓冲区
    private static final BlockingQueue<String> buffer = new LinkedBlockingQueue<>(BUFFER_SIZE);
    
    // 信号量：控制缓冲区中的空位数量（生产者使用）
    private static final Semaphore emptySlots = new Semaphore(BUFFER_SIZE);
    
    // 信号量：控制缓冲区中的数据数量（消费者使用）
    private static final Semaphore fullSlots = new Semaphore(0);
    
    // 互斥锁：保护缓冲区的并发访问
    private static final Semaphore mutex = new Semaphore(1);
    
    public static void main(String[] args) {
        System.out.println("=== 生产者-消费者系统启动 ===");
        System.out.println("缓冲区大小: " + BUFFER_SIZE);
        System.out.println("=============================");
        
        // 创建3个生产者
        for (int i = 1; i <= 3; i++) {
            Thread producer = new Thread(new Producer("生产者-" + i));
            producer.start();
        }
        
        // 创建2个消费者
        for (int i = 1; i <= 2; i++) {
            Thread consumer = new Thread(new Consumer("消费者-" + i));
            consumer.start();
        }
    }
    
    static class Producer implements Runnable {
        private final String producerId;
        private final Random random = new Random();
        private int productCount = 0;
        
        public Producer(String producerId) {
            this.producerId = producerId;
        }
        
        @Override
        public void run() {
            try {
                while (productCount < 5) { // 每个生产者生产5个产品
                    // 生产产品
                    String product = producerId + "-产品-" + (++productCount);
                    
                    // 模拟生产时间
                    Thread.sleep(1000 + random.nextInt(2000));
                    
                    // 等待空位
                    emptySlots.acquire();
                    
                    // 获取互斥锁
                    mutex.acquire();
                    
                    try {
                        // 将产品放入缓冲区
                        buffer.offer(product);
                        System.out.println("📦 " + producerId + " 生产了 " + product + 
                                         " (缓冲区大小: " + buffer.size() + "/" + BUFFER_SIZE + ")");
                    } finally {
                        // 释放互斥锁
                        mutex.release();
                    }
                    
                    // 通知消费者有新产品
                    fullSlots.release();
                }
                
                System.out.println("✅ " + producerId + " 完成生产任务");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("⚠️ " + producerId + " 被中断");
            }
        }
    }
    
    static class Consumer implements Runnable {
        private final String consumerId;
        private final Random random = new Random();
        private int consumedCount = 0;
        
        public Consumer(String consumerId) {
            this.consumerId = consumerId;
        }
        
        @Override
        public void run() {
            try {
                while (consumedCount < 8) { // 每个消费者消费8个产品
                    // 等待产品
                    fullSlots.acquire();
                    
                    // 获取互斥锁
                    mutex.acquire();
                    
                    String product = null;
                    try {
                        // 从缓冲区取出产品
                        product = buffer.poll();
                        if (product != null) {
                            consumedCount++;
                            System.out.println("🛒 " + consumerId + " 消费了 " + product + 
                                             " (缓冲区大小: " + buffer.size() + "/" + BUFFER_SIZE + ")");
                        }
                    } finally {
                        // 释放互斥锁
                        mutex.release();
                    }
                    
                    // 通知生产者有空位
                    emptySlots.release();
                    
                    if (product != null) {
                        // 模拟消费时间
                        Thread.sleep(1500 + random.nextInt(2500));
                    }
                }
                
                System.out.println("✅ " + consumerId + " 完成消费任务");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("⚠️ " + consumerId + " 被中断");
            }
        }
    }
}
